﻿using Furion.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace Furion.EventBus.RabbitMq
{
    public class RabbitMqEventSourceStorer : IEventSourceStorer, IDisposable, ISingleton
    {
        /// <summary>
        /// 中间人
        /// </summary>
        private readonly Channel<IEventSource> _channel;
        private readonly ILogger<RabbitMqEventSourceStorer> _logger;
        private readonly string[] _queueNames;
        private readonly RabbitMqOptions _rabbitMqOptions;
        private ConnectionFactory? _factory;
        private IModel? _mqChannel;

        public RabbitMqEventSourceStorer(IOptions<RabbitMqOptions> rabbitMqOptions, ILogger<RabbitMqEventSourceStorer> logger, IEnumerable<IEventSubscriber>? eventSubscribers)
        {
            _rabbitMqOptions = rabbitMqOptions.Value;
            _logger = logger;

            var boundedChannelOptions = new BoundedChannelOptions(2)
            {
                FullMode = BoundedChannelFullMode.Wait
            };
            _channel = Channel.CreateBounded<IEventSource>(boundedChannelOptions);

            if (eventSubscribers == null)
            {
                _queueNames = Array.Empty<string>();
            }
            else
            {
                _queueNames = eventSubscribers
                    .SelectMany(d =>
                                d.GetType()
                                 .GetMethods(BindingFlags.Public | BindingFlags.Instance | BindingFlags.DeclaredOnly)
                                 .Where(m => m.IsDefined(typeof(EventSubscribeAttribute), false))
                                 .SelectMany(c => c.GetCustomAttributes<EventSubscribeAttribute>(false))
                                 .Select(c => c.EventId.ToString())
                    ).ToArray();
            }

            OpenMqChannel();
        }

        private void OpenMqChannel()
        {
            if (_mqChannel != null && _mqChannel.IsOpen)
            {
                return;
            }

            if (_factory == null)
            {
                _factory = new ConnectionFactory
                {
                    Uri = new Uri(_rabbitMqOptions.ConnectionName ?? throw new ArgumentNullException("RabbitMqOptions.ConnectionName is null"))
                };
            }

            _mqChannel = _factory.CreateConnection().CreateModel();

            _mqChannel.CallbackException += (sender, e) =>
            {
                _logger.LogError(e.Exception, $"RabbitExceptionDetail:{e.Detail}");
            };

            for (int i = 0; i < _queueNames.Length; i++)
            {
                _mqChannel.QueueDeclare(_queueNames[i], false, false, false, null);

                _mqChannel.QueueBind(_queueNames[i], "amq.direct", _queueNames[i]);

                EventingBasicConsumer consumer = new(_mqChannel);

                consumer.Received += (sender, e) =>
                {
                    var payload = Encoding.UTF8.GetString(e.Body.ToArray());
                    IEventSource eventSource = new RabbitMqEventSource(e.RoutingKey, payload);

                    if (_channel.Writer.TryWrite(eventSource))
                    {
                        _mqChannel.BasicAck(e.DeliveryTag, false);
                    }
                };

                _mqChannel.BasicConsume(_queueNames[i], false, consumer);
            }
        }


        public async ValueTask<IEventSource> ReadAsync(CancellationToken cancellationToken)
        {
            if (cancellationToken.IsCancellationRequested && _mqChannel != null && _mqChannel.IsOpen)
            {
                _mqChannel.Close();
            }

            var eventSource = await _channel.Reader.ReadAsync(cancellationToken);
            return eventSource;
        }

        public ValueTask WriteAsync(IEventSource eventSource, CancellationToken cancellationToken)
        {
            if (_mqChannel == null || _mqChannel.IsClosed)
            {
                OpenMqChannel();
            }

            var body = Encoding.UTF8.GetBytes(eventSource.Payload.ToString() ?? "");
            _mqChannel.BasicPublish("amq.direct", eventSource.EventId, body: body);
            return ValueTask.CompletedTask;
        }

        public void Dispose()
        {
            if (_mqChannel != null)
            {
                _mqChannel.Dispose();
            }
        }
    }
}
